package hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class Test {

    public static final String FULL_TABLE_NAME = "wangbixiao:G20200388010132";
    public static final String NAMESPACE = "wangbixiao";



    static byte[] infoCF =  Bytes.toBytes("info");
    static byte[] scoreCF =  Bytes.toBytes("score");

    static byte[] studentIdC = Bytes.toBytes("student_id");
    static byte[] classC = Bytes.toBytes("class");
    static byte[] understandingC = Bytes.toBytes("understanding");
    static byte[] programmingC = Bytes.toBytes("programming");


    public static void main(String[] args) throws IOException {
        Connection connection = createConnection();

        TableName tableName = TableName.valueOf(FULL_TABLE_NAME);

        boolean created = createTable(connection, tableName);

        Table table = connection.getTable(tableName);

        if (created) {
            initData(table);
        }
        insert(table);

        query("wangbixiao", table);

        delete("wangbixiao", table);
    }

    public static Connection createConnection() throws IOException {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.quorum","47.101.206.249:2181,47.101.204.23:2181,47.101.216.12:2181");
        return ConnectionFactory.createConnection(configuration);
    }

    public static boolean createTable(Connection connection, TableName tableName) throws IOException {

        Admin admin = connection.getAdmin();
        try {
            admin.createNamespace(NamespaceDescriptor.create(NAMESPACE).build());
        } catch (Exception e) {
            System.out.println("namespace创建失败，可能已经存在：" + NAMESPACE);
        }

        if (admin.tableExists(tableName)) {
            System.out.println("表已存在：" + tableName.getNameAsString());
            return false;
        }
        // 创建表

        TableDescriptor tableDescriptor = TableDescriptorBuilder
                .newBuilder(tableName)
                .setColumnFamilies(
                        Arrays.asList(
                                ColumnFamilyDescriptorBuilder.of("info"),
                                ColumnFamilyDescriptorBuilder.of("score")
                        ))
                .build();

        admin.createTable(tableDescriptor);
        return true;
    }

    public static void initData(Table table) throws IOException {


        List<Put> putList = new ArrayList<>();
        Put put = new Put(Bytes.toBytes("Tom"));
        put.addColumn(infoCF, studentIdC, Bytes.toBytes("20210000000001"));
        put.addColumn(infoCF, classC, Bytes.toBytes("1"));
        put.addColumn(scoreCF, understandingC, Bytes.toBytes("75"));
        put.addColumn(scoreCF, programmingC, Bytes.toBytes("82"));
        putList.add(put);

        Put put2 = new Put(Bytes.toBytes("Jerry"));
        put2.addColumn(infoCF, studentIdC, Bytes.toBytes("20210000000002"));
        put2.addColumn(infoCF, classC, Bytes.toBytes("1"));
        put2.addColumn(scoreCF, understandingC, Bytes.toBytes("85"));
        put2.addColumn(scoreCF, programmingC, Bytes.toBytes("67"));
        putList.add(put2);

        Put put3 = new Put(Bytes.toBytes("Jack"));
        put3.addColumn(infoCF, studentIdC, Bytes.toBytes("20210000000003"));
        put3.addColumn(infoCF, classC, Bytes.toBytes("2"));
        put3.addColumn(scoreCF, understandingC, Bytes.toBytes("80"));
        put3.addColumn(scoreCF, programmingC, Bytes.toBytes("80"));
        putList.add(put3);

        Put put4 = new Put(Bytes.toBytes("Rose"));
        put4.addColumn(infoCF, studentIdC, Bytes.toBytes("20210000000004"));
        put4.addColumn(infoCF, classC, Bytes.toBytes("2"));
        put4.addColumn(scoreCF, understandingC, Bytes.toBytes("60"));
        put4.addColumn(scoreCF, programmingC, Bytes.toBytes("62"));
        putList.add(put4);


    }

    public static void insert(Table table) throws IOException {
        Put put5 = new Put(Bytes.toBytes("wangbixiao"));
        put5.addColumn(infoCF, studentIdC, Bytes.toBytes("G20200388010132"));
        put5.addColumn(infoCF, classC, Bytes.toBytes("5"));
        put5.addColumn(scoreCF, understandingC, Bytes.toBytes("100"));
        put5.addColumn(scoreCF, programmingC, Bytes.toBytes("100"));

        table.put(put5);
    }

    public static void query(String rowKey, Table table) throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));
        Result result = table.get(get);
        String studentId = Bytes.toString(result.getValue(infoCF, studentIdC));
        String clazz = Bytes.toString(result.getValue(infoCF, classC));
        String understanding = Bytes.toString(result.getValue(scoreCF, understandingC));
        String programming = Bytes.toString(result.getValue(scoreCF, programmingC));

        System.out.println("studentId:" + studentId + ", class:" + clazz
                + ", understanding:" + understanding + ", programming:" + programming);
    }

    public static void delete(String rowKey, Table table) throws IOException {
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        table.delete(delete);
    }
}
